/** @file    SharedMemoryDsm.cpp
 *  @author  Alessandro Polo
 *  @version $Id: SharedMemoryDsm.cpp 2891 2010-08-12 01:58:59Z alex $
 *  @brief
 * File containing methods for the wosh::SharedMemoryDsm class.
 * The header for this class can be found in SharedMemoryDsm.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2010, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/

 #include <core/SharedMemoryDsm.h>

 #include <core/ObjectAllocator.h>
 #include <core/SecurityManager.h>
 #include <core/MessageFilter.h>
 #include <core/Request.h>
 #include <core/Response.h>
 #include <core/Functional.h>


using namespace std;
namespace wosh {

 WOSH_REGISTER(wosh::SharedMemoryDsm, "wosh::SharedMemory", 0.05, _static_SharedMemoryDsm)

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

SharedMemoryDsm::SharedMemoryDsm( const std::string& name, Object* parent )
	: ThreadImpl(parent, name), SharedMemory(name, parent) {

	this->info.strategy = (short)SharedMemory::STRATEGY_EPIDEMIC;

	this->transationsHistorySize = 100;
 }

SharedMemoryDsm::~SharedMemoryDsm() {
 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// FLOW/CONTROL

WRESULT SharedMemoryDsm::init( unsigned long timeout ) {
	WRESULT ret = SharedMemory::init(timeout);
	if ( WFAILED(ret) ) {
		Log(LOG_CRITICAL, ":init() FAILED#%d : base initializer!", ret );
		return ret;
	 }
	Log(LOG_DEBUG, ":init() : Starting intenal thread.." );
	ret = startThread(timeout);
	return ret;
 }

WRESULT SharedMemoryDsm::deinit( unsigned long timeout ) {
	Log(LOG_VERBOSE, ":deinit() : Stopping intenal thread.." );
	WRESULT ret = quitThread(timeout);
	if ( WFAILED(ret) ) {
		Log(LOG_CRITICAL, ":deinit() FAILED#%d : Stopping thread!", ret );
		return ret;
	 }
	ret = SharedMemory::deinit(timeout);
	return ret;
 }

//////////////////////////////////////////////////////////////////////////////////////////////// FLOW/CONTROL
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////// SETUP

WRESULT SharedMemoryDsm::setMode( AGENT_MODE mode, bool raise_update ) {
	if ( mode == SharedMemory::MODE_UNKNOWN ) return WRET_ERR_PARAM;
	Log(LOG_INFO, ":setMode() : Setting to %d", (int)mode );
	//this->agents.transactionBeginWrite();
	//this->info.mode = (short)mode;
	//this->agents.transactionEnd();
	//return WRET_OK;
	return WRET_ERR_ILLEGAL_USE;
 }

WRESULT SharedMemoryDsm::setPropagation( UPDATE_PROPAGATION propagation, bool raise_update ) {
	if ( propagation == SharedMemory::PROPAGATION_UNKNOWN ) return WRET_ERR_PARAM;
	Log(LOG_INFO, ":setPropagation() : Setting to %d", (int)propagation );
	//this->agents.transactionBeginWrite();
	//this->info.propagation = (short)propagation;
	//this->agents.transactionEnd();
	//return WRET_OK;
	return WRET_ERR_ILLEGAL_USE;
 }

/////////////////////////////////////////////////////////////////////////////////////////////////////// SETUP
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////// THREAD CONTROL

bool SharedMemoryDsm::initializingThread() {
	Log(LOG_DEBUG, ":initializingThread() : still nothing to do.." );
	return true;
 }

void SharedMemoryDsm::runThread() {
	Log(LOG_VERBOSE, ":runThread() : Entering transaction loop.." );

	while ( this->running ) {
		ThreadImpl::sleepForMSec(50);
		setThreadAlive();
//continue;
		this->transations.transactionBeginWrite();
		if ( this->transations.size() == 0 ) {
			this->transations.transactionEnd();
			ThreadImpl::sleepForMSec(100);
			continue;
		 }
		Transaction* tr = this->transations.frontItem();
		if ( tr == NULL ) {
			this->transations.erase(this->transations.begin());
			this->transations.transactionEnd();
			continue;
		 }

		if ( tr->state == Transaction::STATE_QUEUED ) {
			// just added? fix internal stuff calling precommit
			WRESULT precommitted = tr->precommit();
			if ( WFAILED(precommitted) ) 
				Log(LOG_CRITICAL, ":runThread() : TR#%"PRId64" Precommit FAILED#%d ", tr->getID(), precommitted );
		 }

		if ( tr->isNACKedBySomeone() ) { // NACKs >= 1
			tr->rollback();
			removeTransaction_(tr);
		 }
		else if ( tr->getResponseOf(this->info.id) == Transaction::NO_REPLY ) {
			Log(LOG_INFO, ":runThread() : ACKing #%"PRId64"", tr->getID() );
			// evaluate the transaction, may we ACK IT?
			tr->getAcksMap()[this->info.id] = Transaction::ACKED;
			// even my own ack is updated from message (just like (n)acks from other agents)
			sendToAgent( new Message(tr->replyACK()), 0 );
		 }
		else if ( tr->isACKedByAll() ) { // tr->getAcksCount() == agents.count()
			tr->commit();
			removeTransaction_(tr);
		 }
		else if ( tr->isTimedOut() ) {
			tr->rollback();
			// all following transactions should be aborted
			removeTransaction_(tr);
		 }
		this->transations.transactionEnd();
	 }
	Log(LOG_VERBOSE, ":runThread() : Exited transaction loop.." );
 }

void SharedMemoryDsm::exitingThread() {
	Log(LOG_INFO, ":exitingThread() : Clearing transaction queue.." );
	this->transations.transactionBeginWrite();
	this->transations.free();
	this->transations.transactionEnd();
 }

////////////////////////////////////////////////////////////////////////////////////////////// THREAD CONTROL
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// TRANSACTIONs

void SharedMemoryDsm::removeTransaction_( Transaction* transaction ) {
	if ( transaction == NULL ) return;
	this->transations.pop(transaction);
	if ( this->transationsHistorySize == 0 )
		delete transaction;
	if ( this->transationsHistorySize <= this->transationsHistory.size() )
		this->transationsHistory.erase(this->transationsHistory.begin(), true);
	this->transationsHistory.push_back(transaction);
 }

WRESULT SharedMemoryDsm::evaluateTransactionRequest( const Message& message ) {
	if ( message.isEmpty() || !message.getContent()->isRequest() ) return WRET_ERR_PARAM;
	const Request* request = message.getContent()->asRequest();
	if ( !request->hasData() || request->getData()->as<Transaction>() == NULL ) {
		Log(LOG_WARNING, ":evaluateTransactionRequest(%"PRId64") : Invalid content!", message.getID() );
		return WRET_ERR_PARAM;
	 }
	// get the transaction (copy)
	Transaction* transaction = request->getData()->clone<Transaction>();
	if ( transaction == NULL ) {
		Log(LOG_CRITICAL, ":evaluateTransactionRequest(%"PRId64") : Invalid cloned content!", message.getID() );
		return WRET_ERR_PARAM;
	 }
	// fix internal fields
	transaction->setParent(this);
	transaction->state = Transaction::STATE_QUEUED;
	this->transations.transactionBeginWrite();
	// ensure it is not already in queue
	ListT<Transaction*> ::ConstIterator it_f = std::find_if(this->transations.begin(), this->transations.end(),
		Functor_Validator<Transaction, int64>(transaction->getID(), &Transaction::getID ) );
	if ( it_f != this->transations.end() && (*it_f) != NULL ) {
		this->transations.transactionEnd();
		Log(LOG_CRITICAL, ":evaluateTransactionRequest(%"PRId64") : Transaction#%"PRId64" is already queued!", message.getID(), transaction->getID() );
		delete transaction;
		return WRET_ERR_PARAM;
	 }
	Log(LOG_INFO, ":evaluateTransactionRequest() : Enqueuing %s..", transaction->toString().c_str() );
	// enqueue it in FIFO queue!
	this->transations.push_back( transaction );
	this->transations.transactionEnd();
	return WRET_OK;
 }

WRESULT SharedMemoryDsm::evaluateTransactionResponse( const Message& message ) {
	if ( message.isEmpty() || !message.getContent()->isResponse() ) return WRET_ERR_PARAM;
	long agent_id = Utilities::fromString<long>(message.getSource().getName(), 0);
	if ( agent_id == 0 ) return WRET_ERR_PARAM;
	const Response* response = message.getContent()->asResponse();
	if ( response == NULL || !response->hasData() ) {
		Log(LOG_WARNING, ":evaluateTransactionResponse(%"PRId64") : Invalid content!", message.getID() );
		return WRET_ERR_PARAM;
	 }
	const Variant* vID = response->getData()->as<Variant>();
	if ( vID == NULL || !vID->isNumeric() ) return WRET_ERR_PARAM;
	int64 transaction_id = vID->toInt64(0);
	if ( transaction_id == 0 ) return WRET_ERR_PARAM;
	this->transations.transactionBeginWrite();
	ListT<Transaction*> ::ConstIterator it_f = std::find_if(this->transations.begin(), this->transations.end(),
		Functor_Validator<Transaction, int64>(transaction_id, &Transaction::getID ) );
	if ( it_f == this->transations.end() || (*it_f) == NULL ) {
		this->transations.transactionEnd();
		Log(LOG_WARNING, ":evaluateTransactionResponse(%"PRId64") : Transaction not found!", transaction_id );
		return WRET_ERR_PARAM;
	 }
	if ( WSUCCEEDED(response->getReturnValue()) ) {
		Log(LOG_VERBOSE, ":evaluateTransactionResponse(%"PRId64") ACKed by %ld", transaction_id, agent_id );
		(*it_f)->getAcksMap()[agent_id] = Transaction::ACKED;
	 }
	else {
		Log(LOG_INFO, ":evaluateTransactionResponse(%"PRId64") NACKed by %ld", transaction_id, agent_id );
		(*it_f)->getAcksMap()[agent_id] = Transaction::NACKED;
	 }
	this->transations.transactionEnd();
	return WRET_OK;
 }

//////////////////////////////////////////////////////////////////////////////////////////////// TRANSACTIONs
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////// BUS EVENTS

void SharedMemoryDsm::busMessage( const Message& message, const Bus* source ) {
	if ( message.isEmpty() ) return;
	SharedMemory::busMessage( message, source );
	if ( !MessageFilter::isMessageTarget(message, Object::getURI() ) ) return;

	if ( message.getContent()->isRequest() ) {
		const Request* request = message.getContent()->asRequest();
		if ( request == NULL ) return;
		if ( request->getMethod() == _SharedMemoryDsm_METHOD_transaction ) {
			// sounds like a new transaction has started..
			evaluateTransactionRequest(message);
		 }
	 }
	else if ( message.getContent()->isResponse() ) {
		const Response* response = message.getContent()->asResponse();
		if ( response == NULL ) return;
		if ( response->getMethod() == _SharedMemoryDsm_METHOD_transaction ) {
			// a transaction was (N)ACKed by someone?
			evaluateTransactionResponse(message);
		 }
	 }
 }

////////////////////////////////////////////////////////////////////////////////////////////////// BUS EVENTS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

}; // namespace wosh
